package com.hex.ds.hdrs.period.job.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.hex.bigdata.hbda.base.Page;
import com.hex.bigdata.hbda.utils.StringUtils;
import com.hex.ds.hdrs.common.constant.Const;
import com.hex.ds.hdrs.common.page.PageData;
import com.hex.ds.hdrs.common.utils.BaseUtil;
import com.hex.ds.hdrs.common.utils.HdrsDateUtil;
import com.hex.ds.hdrs.common.utils.MapUtil;
import com.hex.ds.hdrs.engine.service.HandService;
import com.hex.ds.hdrs.label.po.LabelInfoPo;
import com.hex.ds.hdrs.label.service.ILabelInfoService;
import com.hex.ds.hdrs.period.job.converter.IPeriodJobConverter;
import com.hex.ds.hdrs.period.job.dao.PeriodJobDao;
import com.hex.ds.hdrs.period.job.dto.PeriodJobCopyDto;
import com.hex.ds.hdrs.period.job.dto.PeriodJobDto;
import com.hex.ds.hdrs.period.job.po.PeriodJob;
import com.hex.ds.hdrs.period.job.vo.PeriodJobVo;
import com.hex.ds.hdrs.period.task.converter.IPeriodTaskConverter;
import com.hex.ds.hdrs.period.task.po.PeriodTask;
import com.hex.ds.hdrs.period.task.service.PeriodTaskDefService;
import com.hex.ds.hdrs.period.task.service.PeriodTaskService;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

/**
 * @Package: com.hex.ds.hdrs.period.job.service
 * @ClassName PeriodJobService
 * @Description: 周期作业 服务层
 * @Author: jd.chen
 * @Date 2021/7/8 16:03
 * @Version v1.0
 **/
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy, @Autowired})
public class PeriodJobService {

    private final PeriodJobDao periodJobDao;

    private final PeriodTaskService periodTaskService;

    private final HandService handService;

    private final IPeriodJobConverter converter;

    private final IPeriodTaskConverter periodTaskConverter;

    private final ILabelInfoService labelInfoService;

    private final PeriodTaskDefService periodTaskDefService;

    /*
     * @Method: queryListByPage <br>
     * @Param: [periodJobDto, page] <br>
     * @Return: com.hex.ds.hdrs.common.page.PageData <br>
     * @Description：分页查询<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/4 11:07 <br>
     * @Version： V2.0.2<br>
     */
    public PageData queryListByPage(PeriodJobDto periodJobDto, Page page) {
        List<PeriodJob> periodJobList = periodJobDao.queryListByPage(converter.toPeriodJob(periodJobDto), page);
        for (PeriodJob periodJob : periodJobList) {
            List<LabelInfoPo> labelInfoPoList = labelInfoService.queryListByJobCode(periodJob.getJobCode());
            periodJob.setLabelName(CollectionUtil.join(labelInfoPoList.stream().map(LabelInfoPo::getLabelName).collect(Collectors.toList()), ","));
            periodJob.setTaskStatusCount(getTaskStatus(periodJob));
        }
        return new PageData(periodJobList, page);
    }

    /*
     * @Method: getTaskStatus <br>
     * @Param: [periodJob] <br>
     * @Return: java.lang.String <br>
     * @Description：获取任务实例状态统计<br>
     * @Author： wz.li<br>
     * @Date： 2023/7/7 13:49 <br>
     * @Version： V2.0.2<br>
     */
    private String getTaskStatus(PeriodJob periodJob) {
        List<PeriodTask> periodTaskList = periodTaskService.qryTaskListInJob(periodJob.getJobCode(), periodJob.getBizDate());
        int pending = 0, completed = 0, pendfail = 0;
        for (PeriodTask periodTask : periodTaskList) {
            String taskStatus = periodTask.getTaskStatus().trim();
            if (Const.TASK_TODO_RUN.equals(taskStatus) || Const.TASK_RUNNING.equals(taskStatus) || Const.TASK_STOP.equals(taskStatus)) {
                pending += 1;
            } else if (StrUtil.equals(Const.TASK_SUCCESS, taskStatus) || Const.TASK_HAND_SUCCESS.equals(taskStatus)) {
                completed += 1;
            } else if (Const.TASK_FAIL.equals(taskStatus)) {
                pendfail = 1;
            }
        }
        return pending + "|" + completed + "|" + pendfail;
    }

    /**
     * 查询全量作业
     *
     * @return
     */
    public List<PeriodJob> qryJobList() {
        log.info("查询job list 【PeriodJobService.qryJobList】======》");
        return periodJobDao.qryJobList();
    }

    /**
     * 查询增量作业
     *
     * @param lastQryTs
     * @return
     */
    public List<PeriodJob> qryJobList(String lastQryTs) {
        log.info("查询job list 【PeriodJobService.qryJobList】参数【lastQryTs】:{}",lastQryTs);
        return periodJobDao.qryJobList(lastQryTs);
    }

    /**
     * 查询全量 最新完成作业
     *
     * @return
     */
    public List<PeriodJob> qryLstFnsJobList() {
        log.info("查询 job list 【PeriodJobService.qryLstFnsJobList】");
        return periodJobDao.qryLstFnsJobList();
    }

    /**
     * 查询增量 最新完成作业
     *
     * @param lastQryTs
     * @return
     */
    public List<PeriodJob> qryLstFnsJobList(String lastQryTs) {
        return periodJobDao.qryLstFnsJobList(lastQryTs);
    }

    /**
     * 查询一个业务日期的作业
     *
     * @param jobCode
     * @return
     */
    public PeriodJob qryNextDateJob(String jobCode) {
        Map map = new HashMap(16);
        map.put("jobCode", jobCode);
        return this.periodJobDao.getNextPeriodJob(map);
    }

    /*
     * @Method: queryByJobCodeAndBizDate <br>
     * @Param: [jobCode, bizDate] <br>
     * @Return: com.hex.ds.hdrs.period.job.po.PeriodJob <br>
     * @Description：根据指定的作业编码和业务日期查询对应的作业实例<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 10:04 <br>
     * @Version： V2.0.2<br>
     */
    public PeriodJob queryByJobCodeAndBizDate(String jobCode, String bizDate) {
        PeriodJob periodJob = new PeriodJob().setJobCode(jobCode).setBizDate(bizDate);
        return periodJobDao.queryOne(periodJob);
    }

    /*
     * @Method: queryListByJobCode <br>
     * @Param: [jobCode] <br>
     * @Return: java.util.List<com.hex.ds.hdrs.period.job.po.PeriodJob> <br>
     * @Description：根据指定的作业编码查询对应的作业实例列表<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 11:08 <br>
     * @Version： V2.0.2<br>
     */
    public List<PeriodJob> queryListByJobCode(String jobCode) {
        PeriodJob periodJob = new PeriodJob().setJobCode(jobCode);
        return periodJobDao.queryList(periodJob);
    }

    /**
     * @Method: updateJob
     * @Param: [map]
     * @Return: void
     * @Description： 更新作业
     * @Author： jd.chen
     * @Date： 2021/6/24 13:59
     * @Version： V1.0
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void updateJob(Map map) {
        log.info("【PeriodJobService.updateJob】map:{}", map);
        this.periodJobDao.updateJob(map);
    }

    /**
     * @Method: upJobToToDo
     * @Param: [jobCodeList, bizDate]
     * @Return: void
     * @Description： 批量修改作业状态为待处理
     * @Author： jd.chen
     * @Date： 2021/6/21 23:28
     * @Version： V1.0
     */
    public void upJobToToDo(String jobCode, String bizDate, String upTs) {
        Map paramMap = new HashMap(16);
        paramMap.put("jobCode", jobCode);
        paramMap.put("bizDate", bizDate);
        paramMap.put("jobStatus", Const.JOB_TO_DO);
        paramMap.put("jobStTs", " ");
        paramMap.put("jobEndTs", " ");
        paramMap.put("errorInfo", " ");
        paramMap.put("upTs", upTs);
        log.info("修改作业状态为待处理，作业编码：{}",jobCode);
        this.updateJob(paramMap);
    }

    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public Boolean add(PeriodJob periodJob) {
        if (queryByJobCodeAndBizDate(periodJob.getJobCode(), periodJob.getBizDate()) != null) {
            log.error("作业实例已存在，作业编码: {},业务日期: {}" ,periodJob.getJobCode(), periodJob.getBizDate());
            return true;
        }
        return periodJobDao.insert(periodJob) > 0;
    }

    /**
     * 删除 job
     *
     * @param pkIds
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void delJobInfo(List<String> pkIds) {
        StringBuilder sb = new StringBuilder();
        if (CollectionUtils.isNotEmpty(pkIds)) {
            List<PeriodJob> periodJobs = qryJobInfoByPkIds(pkIds);
            this.periodJobDao.delJobInfo(pkIds);
            for (PeriodJob periodJob : periodJobs) {
                String jobCode = periodJob.getJobCode();
                String bizDate = periodJob.getBizDate();
                if (sb.length() > 0) {
                    sb.append(", ");
                }
                sb.append(jobCode+" "+bizDate);
                periodTaskService.deleteTaskByJob(jobCode, bizDate);
                handService.removeToDoTask(jobCode, bizDate);
                handService.removeToRunJob(jobCode, bizDate, "TASK_DELETE");
                handService.removeToRunningJob(jobCode, bizDate, "TASK_DELETE");
            }
            log.info("删除作业:{}",sb);
        }
    }

    /**
     * @Method: upJobToSuccess
     * @Param: [jobPkId, jobStTs]
     * @Return: void
     * @Description： 更新作业成处理成功
     * @Author： jd.chen
     * @Date： 2021/6/21 20:13
     * @Version： V1.0
     */
    public void upJobToSuccess(String jobPkId, String jobStTs, String upTs) {
        Map map = new HashMap(16);
        map.put("jobStatus", Const.JOB_DO_SUCCESS);
        map.put("jobStTs", StringUtils.defaultString(jobStTs, upTs));
        map.put("jobEndTs", upTs);
        map.put("pkId", jobPkId);
        map.put("upTs", upTs);
        map.put("upUser", BaseUtil.getUserId());
        this.updateJob(map);
    }

    /**
     * @Method: upJobToToDo
     * @Param: [jobPkId]
     * @Return: void
     * @Description： 更新作业为待处理
     * @Author： jd.chen
     * @Date： 2021/6/21 20:16
     * @Version： V1.0
     */
    public void upJobToToDo(String jobPkId, String upTs) {
        Map map = new HashMap(16);
        map.put("jobStatus", Const.JOB_TO_DO);
        map.put("jobStTs", " ");
        map.put("jobEndTs", " ");
        map.put("pkId", jobPkId);
        map.put("upTs", upTs);
        map.put("upUser", BaseUtil.getUserId());
        map.put("errorInfo", " ");
        this.updateJob(map);
    }

    /**
     * 批量修改业务日期
     *
     * @param periodJobVo
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void batchUpdateJobBizDate(PeriodJobVo periodJobVo) {
        log.debug("【PeriodJobService.updatejobsinfo】参数periodJob: {}", periodJobVo);
        if (CollectionUtils.isNotEmpty(periodJobVo.getPkIds())) {
            for (String pkId : periodJobVo.getPkIds()) {
                PeriodJob job = getJobInfoByPkId(pkId);
                if (queryByJobCodeAndBizDate(job.getJobCode(), periodJobVo.getBizDate()) != null) {
                    throw new RuntimeException("作业编码: " + job.getJobCode() + ",业务日期: " + periodJobVo.getBizDate() + " 已经存在不允许修改！");
                }
                updateJobAndTaskByBizDate(job, periodJobVo.getBizDate());
            }
        }
    }

    /**
     * @Method: updateJobAndTaskByBizDate
     * @Param: [job, bizDate]
     * @Return: void
     * @Description: 修改作业和任务实例
     * @Author: ji.tang
     * @Date: 2022/11/24 18:33
     * @Version: V1.0
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void updateJobAndTaskByBizDate(PeriodJob job, String bizDate) {
        //1.修改任务
        updatePeriodTaskByBizDate(job, bizDate);
        //2.修改作业
        updatePeriodJobByBizDate(job, bizDate);
    }

    /**
     * @Method: updatePeriodJobByBizDate
     * @Param: [job, bizDate]
     * @Return: void
     * @Description: 修改作业实例
     * @Author: ji.tang
     * @Date: 2022/11/24 18:37
     * @Version: V1.0
     */
    private void updatePeriodJobByBizDate(PeriodJob job, String bizDate) {
        job.setUpTs(HdrsDateUtil.getCurrentTime());
        job.setUpUser(BaseUtil.getUserId());
        job.setBizDate(bizDate);
        Map map = MapUtil.Entity2Map(job);
        log.info("修改作业实例,作业编码：{},业务日期:{}",job.getJobCode(),job.getBizDate());
        this.periodJobDao.updatePeriodJobByBizDate(map);
    }

    /**
     * @Method: updatePeriodTaskByBizDate
     * @Param: [job, bizDate]
     * @Return: void
     * @Description: 修改任务实例
     * @Author: ji.tang
     * @Date: 2022/11/24 18:38
     * @Version: V1.0
     */
    private void updatePeriodTaskByBizDate(PeriodJob job, String bizDate) {
        Map map = new HashMap();
        map.put("appCode", job.getAppCode());
        map.put("bizDate", job.getBizDate());
        map.put("jobCode", job.getJobCode());
        List<PeriodTask> periodTaskList = this.periodTaskService.qryPeriodTaskList(map);
        log.info("修改任务实例,作业编码：{},业务日期:{}",job.getJobCode(),job.getBizDate());
        this.periodTaskService.batchUpdateTaskBizDate(periodTaskList, bizDate);
    }

    /*
     * @Method: queryListByAppCode <br>
     * @Param: [appCode] <br>
     * @Return: java.util.List<com.hex.ds.hdrs.period.job.po.PeriodJob> <br>
     * @Description：根据应用编码查询指定的周期作业实例<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 10:56 <br>
     * @Version： V2.0.2<br>
     */
    public List<PeriodJob> queryListByAppCode(String appCode) {
        PeriodJob periodJob = new PeriodJob().setAppCode(appCode);
        return periodJobDao.queryList(periodJob);
    }

    public PeriodJob getJobInfoByPkId(String pkId) {
        PeriodJob periodJob = new PeriodJob().setPkId(pkId);
        return periodJobDao.queryOne(periodJob);
    }

    public List<Map> countGroupByStatus(String appCode, String bizDate) {
        log.debug("【PeriodJobService.countGroupByStatus】参数appCode:{}, bizDate:{}", appCode, bizDate);
        Map<String, String> map = new HashMap<>(16);
        map.put("appCode", appCode);
        map.put("bizDate", bizDate);
        return this.periodJobDao.countGroupByStatus(map);
    }

    public List<Map> countGroupByStatus(String appCode) {
        log.debug("【PeriodJobService.countGroupByStatus】参数appCode:{}", appCode);
        Map<String, String> map = new HashMap<>(16);
        map.put("appCode", appCode);
        return this.periodJobDao.countGroupByStatus(map);
    }

    /*
     * @Method: copyPeriodJob <br>
     * @Param: [periodJobInfVo] <br>
     * @Return: void <br>
     * @Description：复制周期作业实例<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 10:07 <br>
     * @Version： V2.0.2<br>
     */
    public void copyPeriodJob(PeriodJobCopyDto periodJobCopyDto) {
        if (CollectionUtils.isNotEmpty(periodJobCopyDto.getDateList()) && CollectionUtils.isNotEmpty(periodJobCopyDto.getPkIds())) {
            List<PeriodJob> periodJobList = qryJobInfoByPkIds(periodJobCopyDto.getPkIds());
            StringBuilder sb = new StringBuilder();
            for (PeriodJob periodJob : periodJobList) {
                if (sb.length() > 0) {
                    sb.append(", ");
                }
                sb.append(periodJob.getJobCode());
                copyPeriodJob(periodJob, periodJobCopyDto.getDateList(), periodJobCopyDto.getJobCopyStatus());
            }
            log.info("复制周期作业实例,作业编码：{}",sb);
        }
    }

    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void chasingDataPeriodJob(PeriodJobCopyDto periodJobCopyDto) {
        if (CollectionUtils.isNotEmpty(periodJobCopyDto.getDateList()) && CollectionUtils.isNotEmpty(periodJobCopyDto.getPkIds())) {
            List<PeriodJob> periodJobList = queryDistinctPeriodJobList(periodJobCopyDto.getPkIds());
            List<String> jobCodeList = periodJobList.stream().map(PeriodJob::getJobCode).collect(Collectors.toList());

            Map<String, List<PeriodTask>> periodTaskMapByJob = queryPeriodTaskByJobIds(periodJobList);
            List<PeriodTask> periodTaskList = periodTaskMapByJob.values().stream().flatMap(List::stream).collect(Collectors.toList());

            List<String> taskCodeList = periodTaskList.stream().map(PeriodTask::getTaskCode).collect(Collectors.toList());
            for (PeriodJob periodJob : periodJobList) {
                if (periodTaskMapByJob.containsKey(periodJob.getJobCode()) && CollectionUtils.isNotEmpty(periodTaskMapByJob.get(periodJob.getJobCode()))) {
                    buildChasingDataPeriodJob(periodJob, periodTaskMapByJob.get(periodJob.getJobCode()),
                            periodJobCopyDto.getJobCopyStatus(), periodJobCopyDto.getDateList(), jobCodeList, taskCodeList);
                }
            }
        }
    }

    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void buildChasingDataPeriodJob(PeriodJob periodJob, List<PeriodTask> periodTaskList, String status,
                                          List<String> bizDateList, List<String> jobCodeList, List<String> taskCodeList) {
        for (String bizDate : bizDateList) {
            String bizDateFormate = HdrsDateUtil.getYYYYMMDDDate(bizDate);
            if (queryByJobCodeAndBizDate(periodJob.getJobCode(), bizDateFormate) == null) {
                // 作业实例复制
                copyPeriodJob(periodJob, bizDate, status);
                // 任务实例复制
                buildChasingDataPeriodTask(periodTaskList, status, bizDateFormate, jobCodeList, taskCodeList, periodJob);
            }
        }
    }

    private void buildChasingDataPeriodTask(List<PeriodTask> periodTaskList, String status, String bizDate,
                                                 List<String> jobCodeList, List<String> taskCodeList, PeriodJob periodJob) {
        List<PeriodTask> existPeriodTaskList = periodTaskDefService.qryTaskByJobCodeBizDate(periodJob);
        StringBuilder sb = new StringBuilder();
        for (PeriodTask periodTask : periodTaskList) {
            if (!periodTaskDefService.checkPeriodTaskExist(periodTask.getTaskCode(),existPeriodTaskList)){
                if (sb.length() > 0) {
                    sb.append(", ");
                }
                sb.append(periodTask.getTaskCode());
                periodTask.setDependJob(getNewDependJob(jobCodeList, periodTask.getDependJob()));
                periodTask.setDependTask(getNewDependTask(taskCodeList, periodTask.getDependTask()));
                periodTaskService.add(periodTaskConverter.copyPeriodTask(periodTask, status, bizDate));
            }
        }
        log.info("任务实例复制,任务编码：{}",sb);
    }

    private String getNewDependJob(List<String> jobCodeList, String oldDependJob) {
        if (StrUtil.isNotBlank(oldDependJob)) {
            List<String> oldDependJobList = new ArrayList<>(Arrays.asList(StrUtil.split(oldDependJob, ",")));
            List<String> intersection = oldDependJobList.stream().filter(jobCodeList::contains).collect(Collectors.toList());
            return StrUtil.join(",", intersection);
        }
        return "";
    }

    private String getNewDependTask(List<String> taskCodeList, String oldDependTask) {
        if (StrUtil.isNotBlank(oldDependTask)) {
            List<String> oldDependTaskList = new ArrayList<>(Arrays.asList(StrUtil.split(oldDependTask, ",")));
            List<String> intersection = oldDependTaskList.stream().filter(taskCodeList::contains).collect(Collectors.toList());
            return StrUtil.join(",", intersection);
        }
        return "";
    }

    private List<PeriodJob> queryDistinctPeriodJobList(List<String> pkIds) {
        List<PeriodJob> periodJobList = qryJobInfoByPkIds(pkIds);
        return periodJobList.stream()
                .collect(Collectors.groupingBy(PeriodJob::getJobCode))
                .values()
                .stream()
                .map(list -> list.get(0))
                .collect(Collectors.toList());
    }

    public Map<String, List<PeriodTask>> queryPeriodTaskByJobIds(List<PeriodJob> periodJobList) {
        Map<String, List<PeriodTask>> periodTaskMap = new HashMap<>();
        for (PeriodJob periodJob : periodJobList) {
            List<PeriodTask> periodTaskList = periodTaskService.qryTaskListInJob(periodJob.getJobCode(), periodJob.getBizDate());
            periodTaskMap.put(periodJob.getJobCode(), periodTaskList);
        }
        return periodTaskMap;
    }

    /*
     * @Method: copyPeriodJob <br>
     * @Param: [periodJob, dateList, jobStatus] <br>
     * @Return: void <br>
     * @Description：复制周期作业实例<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 13:50 <br>
     * @Version： V2.0.2<br>
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void copyPeriodJob(PeriodJob periodJob, List<String> dateList, String jobStatus) {
        try {
            for (String bizDate : dateList) {
                List<PeriodTask> taskList = periodTaskService.qryTaskListInJob(periodJob.getJobCode(), periodJob.getBizDate());
                if (CollectionUtils.isNotEmpty(taskList) && copyPeriodJob(periodJob, bizDate, jobStatus)) {
                    copyTask(taskList, bizDate, jobStatus);
                }
            }
        } catch (Exception e) {
            log.error("复制周期作业实例异常: {}", e.getMessage(), e);
        }
    }

    /*
     * @Method: copyPeriodJob <br>
     * @Param: [periodJob, bizDate, jobStatus] <br>
     * @Return: boolean <br>
     * @Description：复制周期作业实例<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 11:26 <br>
     * @Version： V2.0.2<br>
     */
    private boolean copyPeriodJob(PeriodJob periodJob, String bizDate, String jobStatus) {
        return add(converter.copyPeriodJob(periodJob, jobStatus, HdrsDateUtil.getYYYYMMDDDate(bizDate)));
    }

    /*
     * @Method: copyTask <br>
     * @Param: [taskList, bizDate, taskStatus] <br>
     * @Return: void <br>
     * @Description：复制周期任务实例<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 13:50 <br>
     * @Version： V2.0.2<br>
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void copyTask(List<PeriodTask> taskList, String bizDate, String taskStatus) {
        for (PeriodTask periodTask : taskList) {
            PeriodTask task = periodTaskService.qryPeriodTaskInfo(periodTask.getTaskCode(),HdrsDateUtil.getYYYYMMDDDate(bizDate));
            if (task == null) {
                periodTaskService.add(periodTaskConverter.copyPeriodTask(periodTask, taskStatus, HdrsDateUtil.getYYYYMMDDDate(bizDate)));
            }
        }
    }

    /*
     * @Method: qryJobInfoByPkIds <br>
     * @Param: [pkIds] <br>
     * @Return: java.util.List<com.hex.ds.hdrs.period.job.po.PeriodJob> <br>
     * @Description：根据指定的周期作业实例主键查询对应的实例信息<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/5 11:01 <br>
     * @Version： V2.0.2<br>
     */
    public List<PeriodJob> qryJobInfoByPkIds(List<String> pkIds) {
        PeriodJob periodJob = new PeriodJob().setPkIds(pkIds);
        return periodJobDao.queryList(periodJob);
    }

    /**
     * 查询上游作业数(包含作业状态条件)
     *
     * @param map
     * @return
     */
    public int upDependJobCount(Map map) {
        return periodJobDao.upDependJobCount(map);
    }

    public List<PeriodJob> queryUpDependJob(Map map){
        return periodJobDao.queryUpDependJob(map);
    }

    /**
     * @Method: upDependJobCountNoStatus
     * @Param: [map]
     * @Return: int
     * @Description: 查询上游作业数(不包含作业状态条件)
     * @Author: ji.tang
     * @Date: 2022/11/22 14:41
     * @Version: V1.0
     */
    public int upDependJobCountNoStatus(Map map) {
        return periodJobDao.upDependJobCountNoStatus(map);
    }

    /**
     * @Method: delJobByDate <br>
     * @Param: [map] <br>
     * @Return: void <br>
     * @Description：根据日期删除周期实例数据<br>
     * @Author： gc.zuo<br>
     * @Date： 2021/6/3 16:21 <br>
     * @Version： V1.0<br>
     */
    public void delJobByDate(Map map) {
        periodJobDao.delJobByDate(map);
    }

    /**
     * @Method: reductionJob <br>
     * @Param: [bitchId] <br>
     * @Return: void <br>
     * @Description：还原周期作业实例数据<br>
     * @Author： gc.zuo<br>
     * @Date： 2021/6/3 16:21 <br>
     * @Version： V1.0<br>
     */
    public void reductionJob(String bitchId) {
        periodJobDao.reductionJob(bitchId);
    }

    /**
     * @Method: qryJobStatus <br>
     * @Param: [periodJob] <br>
     * @Return: void <br>
     * @Description：根据作业编码和业务日期查询到作业状态<br>
     * @Author： gc.zuo<br>
     * @Date： 2021/6/22 16:51 <br>
     * @Version： V1.0<br>
     */
    public String qryJobStatus(PeriodJobDto periodJobDto) {
        PeriodJob periodJob = periodJobDao.queryOne(converter.toPeriodJob(periodJobDto));
        return (periodJob != null) ? periodJob.getJobStatus() : "null";
    }

    /**
     * @Method: runAgainOne
     * @Param: [pkIds]
     * @Return: void
     * @Description： 批量重跑单个作业及其下的任务
     * @Author： jd.chen
     * @Date： 2021/6/21 19:44
     * @Version： V1.0
     */
    public void runAgainOne(List<String> pkIds) {
        List<PeriodJob> periodJobList = qryJobInfoByPkIds(pkIds);
        StringBuilder sb = new StringBuilder();
        for (PeriodJob periodJob : periodJobList) {
            if (sb.length() > 0) {
                sb.append(", ");
            }
            sb.append(periodJob.getJobCode());
            runAgainOne(periodJob);
        }
        log.info("批量重跑单个作业及其下的任务,作业编码：{}",sb);
    }

    /*
     * @Method: runAgainOne <br>
     * @Param: [periodJob] <br>
     * @Return: void <br>
     * @Description：重跑指定作业及任务实例<br>
     * @Author： wz.li<br>
     * @Date： 2023/5/16 11:54 <br>
     * @Version： V2.0.2<br>
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void runAgainOne(PeriodJob periodJob) {
        try {
            List<PeriodTask> periodTaskList = periodTaskService.qryTaskListInJob(periodJob.getJobCode(), periodJob.getBizDate());
            if (CollectionUtils.isNotEmpty(periodTaskList)) {
                List<String> taskCodeList = periodTaskList.stream().map(PeriodTask::getTaskCode).collect(Collectors.toList());
                log.info("重跑作业及任务实例，作业编码：{}，业务日期：{}",periodJob.getJobCode(),periodJob.getBizDate());
                upJobToToDo(periodJob.getPkId(), HdrsDateUtil.getCurrentTime());
                periodTaskService.batchUpdateTask(taskCodeList, periodJob.getBizDate());
                handService.removeToRunningJob(periodJob.getJobCode(), periodJob.getBizDate(), "重跑作业/任务");
            }
        } catch (Exception e) {
            log.error("手动重跑作业及任务失败: {}", e.getMessage(), e);
        }
    }

    /**
     * @Method: qryJobDoingButAllTaskFinished
     * @Param: []
     * @Return: java.util.List<com.hex.ds.hdrs.period.job.po.PeriodJob>
     * @Description： 更新异常的作业，任务完成，作业未更新成功的
     * @Author： jd.chen
     * @Date： 2021/7/28 16:39
     * @Version： V1.0
     */
    public List<PeriodJob> qryJobDoingButAllTaskFinished() {
        return this.periodJobDao.qryJobDoingButAllTaskFinished();
    }

    public List<PeriodJob> qryAllTaskHandSuccessJobNotSuccess(){
        return this.periodJobDao.qryAllTaskHandSuccessJobNotSuccess();
    }

    public List<PeriodJob> qryFailJobInNoTaskFail() {
        return this.periodJobDao.qryFailJobInNoTaskFail();
    }
}
